package com.infore.exchange.main.service.impl;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.alibaba.druid.util.JdbcUtils;
import com.baomidou.mybatisplus.service.impl.ServiceImpl;
import com.github.pagehelper.Page;
import com.infore.common.annotation.RecordLog;
import com.infore.common.exception.MyException;
import com.infore.common.model.dto.exchange.JobTaskInfoQuartz;
import com.infore.common.model.dto.exchange.TransformInfoDto;
import com.infore.common.model.entity.exchange.*;
import com.infore.common.request.addupdate.exchange.TTransformInfoSaveRequest;
import com.infore.common.request.query.exchange.TTransformInfoRequest;
import com.infore.common.response.ApiEntityResponse;
import com.infore.common.response.ApiPageResponse;
import com.infore.common.response.ApiResponse;
import com.infore.common.utils.UUIDGenerator;
import com.infore.exchange.main.dao.TDatabaseInfoMapper;
import com.infore.exchange.main.dao.TJobInfoMapper;
import com.infore.exchange.main.dao.TSourceDataInfoMapper;
import com.infore.exchange.main.dao.TTransformInfoMapper;
import com.infore.exchange.main.emums.DataSourceType;
import com.infore.exchange.main.emums.FactorDataEnum;
import com.infore.exchange.main.service.*;
import com.infore.exchange.main.utils.SqlUtils;
import com.infore.screw.core.exception.QueryException;
import com.infore.screw.core.metadata.Column;
import com.infore.screw.core.query.DatabaseQuery;
import com.infore.screw.core.query.DatabaseQueryFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.quartz.CronExpression;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.sql.DataSource;
import java.io.Closeable;
import java.sql.ResultSetMetaData;
import java.sql.Timestamp;
import java.sql.Types;
import java.text.ParseException;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * <p>
 * 数据交换 服务实现类
 * </p>
 *
 * @author lijh
 * @since 2023-02-10
 */
@Slf4j
@RecordLog(descript = "数据交换服务")
@Service
public class TTransformInfoServiceImpl extends ServiceImpl<TTransformInfoMapper, TTransformInfo> implements TTransformInfoService {


    @Autowired
    private  TTransformInfoMapper tTransformInfoMapper;

    @Autowired
    private TDatabaseInfoMapper databaseInfoMapper;

    @Autowired
    private TSourceDataInfoMapper tSourceDataInfoMapper;

    @Autowired
    private TJobInfoMapper jobInfoMapper;

    @Autowired
    private TSourceDataInfoService sourceDataInfoService;

    @Autowired
    private TTargetDataInfoService targetDataInfoService;

    @Autowired
    private TExTransformLogService transformLogService;

    @Autowired
    private TTaskInfoService taskInfoService;

    @Override
    @Transactional(rollbackFor={Exception.class})
    public ApiResponse saveTransform(TTransformInfoSaveRequest request) {
        TTransformInfo transformInfo = request.getTransformInfo();

        //数据来源
        TSourceDataInfo sourceDataInfo = request.getSourceDataInfo();
        sourceDataInfo.setSourceId(UUIDGenerator.generator());
        sourceDataInfoService.insert(sourceDataInfo);

        if("tableToTable".equalsIgnoreCase(transformInfo.getTransformType())){
            String sourceSql="";
            TDatabaseInfo sourceDatabase = databaseInfoMapper.selectById(sourceDataInfo.getDatabaseId());
            if("ORACLE".equalsIgnoreCase(sourceDatabase.getDatabaseType())){
                sourceSql= StrUtil.format("select 1 from {}.{} where {} and ROWNUM=1",
                        sourceDatabase.getUserName(),sourceDataInfo.getDataTable(),StringUtils.isEmpty(sourceDataInfo.getWhereCondition())? "1=1" :sourceDataInfo.getWhereCondition()) ;
            }else if("MYSQL".equalsIgnoreCase(sourceDatabase.getDatabaseType())){
                sourceSql=StrUtil.format("select 1 from {}.{} where {} limit 1",
                        sourceDatabase.getDatabaseInstanceName(),sourceDataInfo.getDataTable(),StringUtils.isEmpty(sourceDataInfo.getWhereCondition())? "1=1" :sourceDataInfo.getWhereCondition());
            }
            DataSource dataSource = SqlUtils.getDataSourceByDatabaseInfo(sourceDatabase,null,DataSourceType.DRUID.getType());
            try {
                boolean exeOk =  SqlUtils.checkQuerySql(dataSource, sourceSql);
                if(!exeOk){
                    return ApiResponse.error("500","whereCondition参数格式错误",sourceSql);
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                log.info("最后关闭连接");
                JdbcUtils.close((Closeable) dataSource);
            }
        }else if("httpToTable".equalsIgnoreCase(transformInfo.getTransformType())){
            String url = sourceDataInfo.getUrl();
            String urlMethod = sourceDataInfo.getUrlMethod();
            String incrementType = sourceDataInfo.getIncrementType();
            String incrementColumnStr = sourceDataInfo.getIncrementColumn();
            String[] sourceFields = sourceDataInfo.getDataFields().split(",");
            //参数
            HashMap<String, Object> paramMap = new HashMap<>();
            String responseJsonStr="";
            if(StringUtils.isNotEmpty(incrementColumnStr) && StringUtils.isNotEmpty(incrementType) ){
                //开始时间，结束时间用","分隔
                String[] incrementCols = incrementColumnStr.split(",");
                if(incrementCols.length!=2){
                    throw new MyException("500", "增量同步需要提供开始值和结束值");
                }
                String startSyncTime =null;
                String endLastupdate = null;
                if("date".equals(incrementType)){
                    //默认开始时间
                    startSyncTime = "1970-01-01";
                    //默认截止时间
                    endLastupdate = DateUtil.formatDate(new Date())  ;
                }else  if("dateTime".equals(incrementType)){
                    //默认开始时间
                    startSyncTime = "1970-01-01 00:00:00";
                    //默认截止时间
                    endLastupdate = DateUtil.formatDate(new Date())  ;
                }
                paramMap.put(incrementCols[0], startSyncTime);
                paramMap.put(incrementCols[1], endLastupdate);
            }
            try {
                if("GET".equalsIgnoreCase(urlMethod) ){
                    responseJsonStr = HttpUtil.get(url, paramMap);
                }else {
                    responseJsonStr = HttpUtil.post(url,paramMap);
                }
            } catch (Exception e) {
                e.printStackTrace();
                return ApiResponse.error("500","接口请求异常："+url,e.getMessage());
            }
            JSONObject responseJson = JSONUtil.parseObj(responseJsonStr);
            JSONArray jsonArray = responseJson.getJSONArray("data");
            if(jsonArray==null){
                jsonArray = responseJson.getJSONArray("list");
            }
            if (jsonArray == null) {
                jsonArray = responseJson.getJSONArray("result");
            }
            if (jsonArray == null) {
                jsonArray = responseJson.getJSONArray("records");
            }

           if(jsonArray!=null && jsonArray.size()>0) {
               JSONObject jsonObject = jsonArray.getJSONObject(0);
               for (int j = 0; j < sourceFields.length; j++) {
                   Object val = jsonObject.get(sourceFields[j]);
                   if(val==null){
                       return ApiResponse.error("500","接口返回的结果没有列："+sourceFields[j]);
                   }
               }
           }
        }else{
            return ApiResponse.error("500","不支持的同步类型\t"+transformInfo.getTransformType());
        }
        //数据去向
        TTargetDataInfo targetDataInfo = request.getTargetDataInfo();
        targetDataInfo.setTargetId(UUIDGenerator.generator());
        targetDataInfoService.insert(targetDataInfo);

        //数据交换基础信息
        transformInfo.setTransformId(UUIDGenerator.generator());
        transformInfo.setSourceId(sourceDataInfo.getSourceId());
        transformInfo.setTargetId(targetDataInfo.getTargetId());
        transformInfo.setState(1);
        transformInfo.setCreateTime(new Date());
        transformInfo.setUpdateTime(new Date());
        this.insert(transformInfo);
        return ApiResponse.ok();
    }

    @Override
    @Transactional(rollbackFor={Exception.class})
    public ApiResponse delTransform(String id) {
        TTransformInfo tTransformInfo = this.selectById(id);
        sourceDataInfoService.deleteById(tTransformInfo.getSourceId());
        targetDataInfoService.deleteById(tTransformInfo.getTargetId());
        this.deleteById(id);
        return ApiResponse.ok("200",id);
    }


    @Override
    public ApiPageResponse<TransformInfoDto> queryTransformInfo(TTransformInfoRequest request, Page page){
        List<TransformInfoDto> list = tTransformInfoMapper.queryTransformInfo(request);
        return  ApiPageResponse.ok(list,page);

    }


    @Override
    public ApiEntityResponse queryJobAndTaskStatistical(){
        return ApiEntityResponse.ok(tTransformInfoMapper.queryJobAndTaskStatistical());

    }

    /**
     * 注解中如果不配置rollbackFor属性,那么事物只会在遇到RuntimeException的时候才会回滚,加上rollbackFor=Exception.class,可以让事物在遇到非运行时异常时也回滚
     * 下面方法中的DataSource是自己创建的，不是用@Transactional进行事务处理
     * @param jobTask
     */
//    @Transactional(rollbackFor={RuntimeException.class, Exception.class})
    @Override
    @Async
    public void rdbmsDataExchang (String schedulId,JobTaskInfoQuartz jobTask) {

        try {
            Date taskStartTime = new Date();

            TransformInfoDto dto= tTransformInfoMapper.queryTransformInfoById(jobTask.getTransformParams());
            if(dto==null){
                throw new MyException("500","查询不到数据交换的对象\ttransformId="+jobTask.getTransformParams());
            }
            log.info("同步服务名=="+dto.getTransformName());
            CronExpression cronExpression = null;
            try {
                cronExpression = new CronExpression(jobTask.getCronSchedule());
            } catch (ParseException e) {
                e.printStackTrace();
                throw new MyException("500","cron表达式格式异常："+jobTask.getCronSchedule(),e.getMessage());
            }
            Date nextDate = cronExpression.getTimeAfter(new Date());
            jobInfoMapper.updateExecTime(jobTask.getJobId(),new Date(),nextDate);
            String[] sourceFields = dto.getSourceFields().split(",");
            String[] targetFields = dto.getTargetFields().split(",");
            if(sourceFields.length!=targetFields.length){
                String error = "来源列和目标列数量不一致！来源：" + sourceFields.length + " 目标：" + targetFields.length;
                throw new MyException("500",error);
            }

            TDatabaseInfo sourceDatabase = databaseInfoMapper.selectById(dto.getSourceDatabaseId());
            TDatabaseInfo targetDatabase = databaseInfoMapper.selectById(dto.getTargetDatabaseId());


            DataSource dataSource1 = SqlUtils.getDataSourceByDatabaseInfo(sourceDatabase,null,DataSourceType.DRUID.getType());
            DataSource dataSource2 = SqlUtils.getDataSourceByDatabaseInfo(targetDatabase,null,DataSourceType.DRUID.getType());

            //预插入的数据行
            AtomicInteger dataNumber = new AtomicInteger();
            //插入失败的数据行
            AtomicInteger dirtyDataNumber = new AtomicInteger();

            //增量更新-下次开始值
            String nextStartValue = null;
            String querySql = null;
            String deleteSql = null;
            try {

                JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource1);
                JdbcTemplate jdbcTemplate2 = new JdbcTemplate(dataSource2);

                querySql = String.format("select %s from %s ", String.join(",", sourceFields), dto.getSourceTable());
    //            deleteSql = String.format("delete from %s ",dto.getTargetTable());
                if(StringUtils.isNotEmpty(dto.getSourceWhere())){
                    querySql = querySql + " where " + dto.getSourceWhere();
                }
                //增量同步
                if(StringUtils.isNotEmpty(dto.getSourceIncrementCol()) && StringUtils.isNotEmpty(dto.getSourceIncrementType()) ){
                    log.info("数据增量同步\t"+sourceDatabase.getDatabaseName()+"."+dto.getSourceTable()+"-->>>"+targetDatabase.getDatabaseName()+"."+dto.getTargetTable());
                    if("time".equals(dto.getSourceIncrementType())){
                        //默认开始时间
                        Timestamp startSyncTime = new Timestamp(DateUtils.parseDate("1970-01-01 00:00:00", "yyyy-MM-dd HH:mm:ss").getTime());
                        //默认截止时间
                        Timestamp endLastupdate = new Timestamp(System.currentTimeMillis());
                        if (StringUtils.isNotEmpty(dto.getSourceIncrementStartValue())) {
                            Date startDate = null;
                            if(dto.getSourceIncrementBeforeMinutes()!=null){
                                startDate = DateUtil.offsetMinute(DateUtils.parseDate(dto.getSourceIncrementStartValue(), "yyyy-MM-dd HH:mm:ss"),dto.getSourceIncrementBeforeMinutes());
                            }else{
                                startDate = DateUtils.parseDate(dto.getSourceIncrementStartValue(), "yyyy-MM-dd HH:mm:ss");

                            }
                            startSyncTime = new Timestamp(startDate.getTime());
                        }
                        if (StringUtils.isNotEmpty(dto.getSourceIncrementEndValue())) {
                            endLastupdate = new Timestamp(DateUtils.parseDate(dto.getSourceIncrementEndValue(), "yyyy-MM-dd HH:mm:ss").getTime());
                        }

                        nextStartValue = DateUtil.format(endLastupdate,"yyyy-MM-dd HH:mm:ss");
                        //来源表查询条件
                        String queryWhere = SqlUtils.getDataxWhereSql(sourceDatabase.getDatabaseType(), dto.getSourceIncrementCol(), startSyncTime, endLastupdate);
                        querySql = querySql + (querySql.contains("where") ? " and " : " where ") + queryWhere;


                        //目标表删除条件
    //                    String deleteWhere = SqlUtils.getDataxWhereSql(targetDatabase.getDatabaseType(), dto.getTargetIncrementCol(), startSyncTime, endLastupdate);
    //                    deleteSql = deleteSql+ "where "+deleteWhere;

                    }else if("long".equals(dto.getSourceIncrementType())){
                        Long startLong = 0L;
                        Long endLong = 0L;
                        String queryTableIncremMaxValueSql = String.format("select MAX(%s) as \"maxValue\" from %s ",dto.getSourceIncrementCol(), dto.getSourceTable());
                        Object maxValue = jdbcTemplate.queryForMap(queryTableIncremMaxValueSql).get("maxValue");
                        if(maxValue!=null){
                            endLong = Long.parseLong(maxValue.toString());
                        }
                        if(StringUtils.isNotEmpty(dto.getSourceIncrementStartValue())){
                            startLong = Long.valueOf(dto.getSourceIncrementStartValue());
                        }
                        if(StringUtils.isNotEmpty(dto.getSourceIncrementEndValue())){
                            endLong = Long.valueOf(dto.getSourceIncrementEndValue());
                        }
                        nextStartValue = String.valueOf(endLong);

                        //来源表查询条件
                        String queryWhere = SqlUtils.getDataxWhereSql(dto.getSourceIncrementCol(), startLong ,endLong );
                        querySql = querySql + (querySql.contains("where") ? " and " : " where ") + queryWhere;

                        //目标表删除条件
    //                    String deleteWhere = SqlUtils.getDataxWhereSql(dto.getTargetIncrementCol(),  startLong, endLong );
    //                    deleteSql = deleteSql+ "where "+ deleteWhere;
                    }
                }

                //查询需要同步的数据量
                String queryCountSql = String.format("select count(1) as \"num\" from %s where %s",dto.getSourceTable(),SqlUtils.getWhereCondition(querySql));
                Object num = SqlUtils.executeQuerySql(dataSource1, queryCountSql).get(0).get("num");
                log.info("源数据表需要同步的数据："+num);
                if(Long.valueOf(num.toString())==0){
    //                return ApiResponse.ok("200","源数据表需要同步的数据："+num+"\n"+queryCountSql);
                    //插入日志
                    insertTransformLog(schedulId,jobTask, dto,taskStartTime, dataNumber, dirtyDataNumber);
                    log.info("源数据表需要同步的数据："+num+"\t"+queryCountSql);
                    return;
                }
                //删除数据
    //            boolean delOK = SqlUtils.executeSql(dataSource2, deleteSql);
    //            if(!delOK){
    //                throw new MyException("500","目标表数据删除失败");
    //            }
    //            String insertSql = String.format("INSERT INTO %s (%s) VALUES (%s)", dto.getTargetTable(), String.join(",", targetFields),
    //                    StringUtils.repeat("?", ",", targetFields.length));


                StringBuilder insertOrUpdateSqlSb=new StringBuilder();
                DatabaseQuery query = new DatabaseQueryFactory(dataSource2).newInstance();
                String primaryKey = query.getPrimaryKeys(dto.getTargetTable()).get(0).getColumnName();
                if("ORACLE".equalsIgnoreCase(targetDatabase.getDatabaseType())
                        || "DM".equalsIgnoreCase(targetDatabase.getDatabaseType())){
                    insertOrUpdateSqlSb= SqlUtils.buildOracleSaveOrUpdateSql(dto.getTargetTable(),primaryKey,targetFields);
                }else if("MYSQL".equalsIgnoreCase(targetDatabase.getDatabaseType())){
                    insertOrUpdateSqlSb= SqlUtils.buildMysqlSaveOrUpdateSql(dto.getTargetTable(),targetFields);
                }
//                log.info("primaryKey=="+primaryKey);
                String insertSql =insertOrUpdateSqlSb.toString();
//                log.info("insertSql="+insertSql);

                List batchUpdateData = new ArrayList();
                jdbcTemplate.query(querySql, rs -> {
                    ResultSetMetaData rsMetaData = rs.getMetaData();
                    while (rs.next()) {
                        //主键列在dataObjects中的下标
                        int pkIndex = 0;
                        Object[] dataObjects = new Object[targetFields.length];
                        for (int i = 0; i < targetFields.length; i++) {
                            String columnName = rsMetaData.getColumnName(i + 1);
                            if(columnName.equalsIgnoreCase(primaryKey)){
                                pkIndex = i;
                            }
                            dataObjects[i] = rs.getObject(i + 1);
                            switch (rsMetaData.getColumnType(i + 1)) {
                                case Types.CHAR:
                                case Types.NCHAR:
                                case Types.VARCHAR:
                                case Types.LONGVARCHAR:
                                case Types.NVARCHAR:
                                case Types.LONGNVARCHAR:
                                    dataObjects[i] = rs.getString(i + 1);
                                    //把因子转成因子编码
                                    if(columnName.equalsIgnoreCase("PROJECT_CODE")){
                                        dataObjects[i] = FactorDataEnum.getFactorToCode(dataObjects[i]==null?"":dataObjects[i].toString()) ;
                                    }
                                    break;
                                case Types.CLOB:
                                case Types.NCLOB:
                                    dataObjects[i] = rs.getString(i + 1);
                                    break;
                                case Types.SMALLINT:
                                case Types.TINYINT:
                                case Types.INTEGER:
                                case Types.BIGINT:
                                    try {
                                        dataObjects[i] = rs.getBigDecimal(i + 1);
                                    } catch (Exception e) {
                                        log.warn("获取数据异常,列:{} ;类型:{}", i + 1, rsMetaData.getColumnType(i + 1));
                                        dataObjects[i] = rs.getString(i + 1);
                                    }
                                    break;
                                case Types.NUMERIC:
                                case Types.DECIMAL:
                                    try {
                                        dataObjects[i] = rs.getBigDecimal(i + 1);
                                    } catch (Exception e) {
                                        log.warn("获取数据异常,列:{} ;类型:{}", i + 1, rsMetaData.getColumnType(i + 1));
                                        dataObjects[i] = rs.getString(i + 1);
                                    }
                                    break;
                                case Types.FLOAT:
                                case Types.REAL:
                                case Types.DOUBLE:
                                    try {
                                        dataObjects[i] = rs.getBigDecimal(i + 1);
                                    } catch (Exception e) {
                                        log.warn("获取数据异常,列:{} ;类型:{}", i + 1, rsMetaData.getColumnType(i + 1));
                                        dataObjects[i] = rs.getString(i + 1);
                                    }
                                    break;
                                case Types.TIME:
                                    dataObjects[i] = rs.getTime(i + 1);
                                    break;
                                case Types.DATE:
                                    if (rsMetaData.getColumnTypeName(i + 1).equalsIgnoreCase("year")) {
                                        dataObjects[i] = rs.getInt(i + 1);
                                    } else {
                                        dataObjects[i] = rs.getDate(i + 1);
                                    }
                                    break;
                                case Types.TIMESTAMP:
                                    dataObjects[i] = rs.getTimestamp(i + 1);
                                    break;
                                case Types.BINARY:
                                case Types.VARBINARY:
                                case Types.BLOB:
                                case Types.LONGVARBINARY:
                                    dataObjects[i] = rs.getBytes(i + 1);
                                    break;
                                default:
                                    dataObjects[i] = rs.getObject(i + 1);
                            }
                        }
    //                    log.info("主键下标=="+pkIndex+"\t主键值="+dataObjects[pkIndex]);
                        /**
                         * 给update的?填值
                         */
                        //把数组转换成ArrayList
                        ArrayList<Object> list = new ArrayList<>(Arrays.asList(dataObjects));
                        if("ORACLE".equalsIgnoreCase(targetDatabase.getDatabaseType())
                                || "DM".equalsIgnoreCase(targetDatabase.getDatabaseType())){
                            //首位添加元素
                            list.add(0, dataObjects[pkIndex]);
                            for (int n = 0; n < dataObjects.length; n++) {
                                //修改的时候不能修改主键
                                if(n==pkIndex){
                                    continue;
                                }
                                list.add(dataObjects[n]);
                            }
                        }else{
                            for (int i = 0; i < dataObjects.length; i++) {
                                list.add(dataObjects[i]);
                            }
                        }

                        //把ArrayList转换成数组
                        dataObjects = list.stream().toArray();

                        dataNumber.getAndIncrement();
                        batchUpdateData.add(dataObjects);
                        if (rs.getRow() % 5000 == 0) {
                            SqlUtils.doBatchInsert(insertSql, batchUpdateData, jdbcTemplate2, dirtyDataNumber);
                            batchUpdateData.clear();
                        }
                    }
                    return null;
                });
                if (!batchUpdateData.isEmpty()) {
                    SqlUtils.doBatchInsert(insertSql, batchUpdateData, jdbcTemplate2, dirtyDataNumber);
                }
            } catch (ParseException e) {
                e.printStackTrace();
                throw new MyException("500","数据同步异常",e.getMessage());
            }finally {
                //关闭数据库连接
                log.info("关闭数据源dataSource1,dataSource2");
                JdbcUtils.close((Closeable) dataSource1);
                JdbcUtils.close((Closeable) dataSource2);

            }
            log.info("同步数据完成,需要同步数量：" + dataNumber.get() + " 失败数量：" + dirtyDataNumber.get());
            //插入日志
            insertTransformLog(schedulId,jobTask,dto,taskStartTime, dataNumber, dirtyDataNumber);
            if(dirtyDataNumber.get()>0){
                throw new MyException("500","数据插入失败，失败数量："+ dirtyDataNumber.get());
            }else{
                if(StringUtils.isNotEmpty(nextStartValue)){
                    try {
                        tSourceDataInfoMapper.updateStartValueById(dto.getSourceId(),nextStartValue);
                    } catch (Exception e) {
                        e.printStackTrace();
                        throw new MyException("500","t_ex_source_data_info表更新增量值异常",e.getMessage());
                    }
                }
            }
        } catch (MyException e) {
            e.printStackTrace();
        } catch (DataAccessException e) {
            e.printStackTrace();
        } catch (NumberFormatException e) {
            e.printStackTrace();
        } catch (QueryException e) {
            e.printStackTrace();
        } finally {
            taskInfoService.updateTaskRunningStateById(jobTask.getTaskId(),0);
        }
    }


    /**
    * @Deacription http数据同步至表
    * @MethodName httpDataExchang
    * @param schedulId
    * @param jobTask
    * @return
    * @Author lijianhua
    * @Date 2023/3/13 9:45
    **/
    @Override
//    @Async
    public void httpDataExchang (String schedulId,JobTaskInfoQuartz jobTask) {
        try {
            Date taskStartTime = new Date();
            TransformInfoDto dto= tTransformInfoMapper.queryTransformInfoById(jobTask.getTransformParams());
            if(dto==null){
                throw new MyException("500","查询不到数据交换的对象\ttransformId="+jobTask.getTransformParams());
            }
            CronExpression cronExpression = null;
            try {
                cronExpression = new CronExpression(jobTask.getCronSchedule());
            } catch (ParseException e) {
                e.printStackTrace();
                throw new MyException("500","cron表达式格式异常："+jobTask.getCronSchedule(),e.getMessage());
            }
            /**
             * 下次执行时间
             */
            Date nextDate = cronExpression.getTimeAfter(new Date());
            jobInfoMapper.updateExecTime(jobTask.getJobId(),new Date(),nextDate);
            String[] sourceFields = dto.getSourceFields().split(",");
            String[] targetFields = dto.getTargetFields().split(",");
            if(sourceFields.length!=targetFields.length){
                String error = "来源列和目标列数量不一致！来源：" + sourceFields.length + " 目标：" + targetFields.length;
                throw new MyException("500",error);
            }
            TDatabaseInfo targetDatabase = databaseInfoMapper.selectById(dto.getTargetDatabaseId());
            DataSource dataSource = SqlUtils.getDataSourceByDatabaseInfo(targetDatabase,null,DataSourceType.DRUID.getType());
            //预插入的数据行
            AtomicInteger dataNumber = new AtomicInteger();
            //插入失败的数据行
            AtomicInteger dirtyDataNumber = new AtomicInteger();
            //增量更新-下次开始值
            String nextStartValue = null;
            try {
                String responseJsonStr = null;
                String sourceUrl = dto.getSourceUrl();
                JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
                //增量同步
                if(StringUtils.isNotEmpty(dto.getSourceIncrementCol()) && StringUtils.isNotEmpty(dto.getSourceIncrementType()) ){
                    log.info("数据增量同步\t"+sourceUrl+"-->>>"+targetDatabase.getDatabaseName()+"."+dto.getTargetTable());

                    HashMap<String, Object> paramMap = new HashMap<>();
                    //开始时间，结束时间用","分隔
                    String sourceIncrementCol = dto.getSourceIncrementCol();
                    String[] incrementCols = sourceIncrementCol.split(",");
                    if(incrementCols.length!=2){
                        throw new MyException("500", "增量同步需要提供开始值和结束值");
                    }
                    String startSyncTime =null;
                    String endLastupdate = null;
                    if("date".equals(dto.getSourceIncrementType())){
                        //默认开始时间
                        startSyncTime = "1970-01-01";
                        //默认截止时间
                        endLastupdate = DateUtil.formatDate(new Date())  ;
                    }else  if("dateTime".equals(dto.getSourceIncrementType())){
                        //默认开始时间
                        startSyncTime = "1970-01-01 00:00:00";
                        //默认截止时间
                        endLastupdate = DateUtil.formatDate(new Date())  ;
                    }
                    if (StringUtils.isNotEmpty(dto.getSourceIncrementStartValue())) {
                        startSyncTime = dto.getSourceIncrementStartValue();
                    }
                    if (StringUtils.isNotEmpty(dto.getSourceIncrementEndValue())) {
                        endLastupdate = dto.getSourceIncrementEndValue();
                    }
                    nextStartValue = endLastupdate;
                    paramMap.put(incrementCols[0], startSyncTime);
                    paramMap.put(incrementCols[1], endLastupdate);

                    if("GET".equalsIgnoreCase(dto.getSourceUrlMethod()) ){
                         responseJsonStr = HttpUtil.get(sourceUrl, paramMap);
                    }else {
                        responseJsonStr = HttpUtil.post(sourceUrl,paramMap);
                    }
                }else {
                    log.info("全量同步\t"+sourceUrl+"-->>>"+targetDatabase.getDatabaseName()+"."+dto.getTargetTable());
                    if("GET".equalsIgnoreCase(dto.getSourceUrlMethod()) ){
                        responseJsonStr = HttpUtil.get(sourceUrl);
                    }else {
                        responseJsonStr = HttpUtil.post(sourceUrl,"");
                    }
                }
                JSONObject responseJson = JSONUtil.parseObj(responseJsonStr);
                JSONArray jsonArray = responseJson.getJSONArray("data");
                if(jsonArray==null){
                    jsonArray = responseJson.getJSONArray("list");
                }
                if (jsonArray == null) {
                    jsonArray = responseJson.getJSONArray("result");
                }
                if (jsonArray == null) {
                    jsonArray = responseJson.getJSONArray("records");
                }
                if (jsonArray == null || jsonArray.size()==0) {
                    //插入日志
                    insertTransformLog(schedulId,jobTask, dto,taskStartTime, dataNumber, dirtyDataNumber);
                    log.info("接口需要同步的数据："+jsonArray.size()+"\t"+sourceUrl);
                    return;
                }
                dataNumber.set(jsonArray.size());
                StringBuilder insertOrUpdateSqlSb=new StringBuilder();
                DatabaseQuery query = new DatabaseQueryFactory(dataSource).newInstance();
                String primaryKey = query.getPrimaryKeys(dto.getTargetTable()).get(0).getColumnName();
                List<? extends Column> tableColumns = query.getTableColumns(dto.getTargetTable());

                //查询目标列的类型
                int pkIndex=0;
                String[] targetFieldsColumnType = new String[targetFields.length];
                for (int i = 0; i < targetFields.length; i++) {
                    if(targetFields[i].equalsIgnoreCase(primaryKey)){
                        pkIndex = i;
                    }
                    for (Column column:tableColumns) {
                        String columnName = column.getColumnName();
                        String columnType = column.getColumnType();
                        if (targetFields[i].equalsIgnoreCase(columnName)){
                            targetFieldsColumnType[i] = columnType;
                            break;
                        }
                    }
                }


                if("ORACLE".equalsIgnoreCase(targetDatabase.getDatabaseType())
                        || "DM".equalsIgnoreCase(targetDatabase.getDatabaseType()) ){
                    insertOrUpdateSqlSb= SqlUtils.buildOracleSaveOrUpdateSql(dto.getTargetTable(),primaryKey,targetFields);
                }else if("MYSQL".equalsIgnoreCase(targetDatabase.getDatabaseType())){
                    insertOrUpdateSqlSb= SqlUtils.buildMysqlSaveOrUpdateSql(dto.getTargetTable(),targetFields);
                }
                String insertSql =insertOrUpdateSqlSb.toString();
                log.info("insertSql="+insertSql);
                List batchUpdateData = new ArrayList();
                for (int i = 0; i < jsonArray.size(); i++) {
                    Object[] dataObjects = new Object[targetFields.length];
                    JSONObject jsonObject = jsonArray.getJSONObject(i);
                    for (int j = 0; j < sourceFields.length; j++) {
                        String targetColumnType = targetFieldsColumnType[j];
                        if(targetColumnType.equalsIgnoreCase("date")
                                || targetColumnType.equalsIgnoreCase("datetime")
                                || targetColumnType.equalsIgnoreCase("timestamp")){
    //                        log.info("当前目标列为日期类型,需要转换数据格式\t"+sourceFields[j]);
                            Date date = DateUtil.parse(jsonObject.get(sourceFields[j]).toString(), "yyyy-MM-dd HH:mm:ss");
                            dataObjects[j] = date;
                        }else{
                            dataObjects[j] = jsonObject.get(sourceFields[j]);
                        }
                    }
                    /**
                     * 给update的?填值
                     */
                    //把数组转换成ArrayList
                    ArrayList<Object> list = new ArrayList<>(Arrays.asList(dataObjects));
                    if("ORACLE".equalsIgnoreCase(targetDatabase.getDatabaseType())
                            || "DM".equalsIgnoreCase(targetDatabase.getDatabaseType())){
                        //首位添加元素
                        list.add(0, dataObjects[pkIndex]);
                        for (int n = 0; n < dataObjects.length; n++) {
                            //修改的时候不能修改主键
                            if(n==pkIndex){
                                continue;
                            }
                            list.add(dataObjects[n]);
                        }
                    }else{
                        for (int k = 0; k < dataObjects.length; k++) {
                            list.add(dataObjects[k]);
                        }
                    }
                    //把ArrayList转换成数组
                    dataObjects = list.stream().toArray();
                    batchUpdateData.add(dataObjects);
                    if (batchUpdateData.size() % 5000 == 0) {
                        SqlUtils.doBatchInsert(insertSql, batchUpdateData, jdbcTemplate, dirtyDataNumber);
                        batchUpdateData.clear();
                    }
                }
                if (!batchUpdateData.isEmpty()) {
                    SqlUtils.doBatchInsert(insertSql, batchUpdateData, jdbcTemplate, dirtyDataNumber);
                }
            } catch (Exception e) {
                e.printStackTrace();
                throw new MyException("500","数据同步异常",e.getMessage());
            }finally {
                //关闭数据库连接
                log.info("dataSource");
                JdbcUtils.close((Closeable) dataSource);
            }
            log.info("同步数据完成,需要同步数量：" + dataNumber.get() + " 失败数量：" + dirtyDataNumber.get());
            //插入日志
            insertTransformLog(schedulId,jobTask,dto,taskStartTime, dataNumber, dirtyDataNumber);
            if(dirtyDataNumber.get()>0){
                throw new MyException("500","数据插入失败，失败数量："+ dirtyDataNumber.get());
            }else{
                if(StringUtils.isNotEmpty(nextStartValue)){
                    try {
                        tSourceDataInfoMapper.updateStartValueById(dto.getSourceId(),nextStartValue);
                    } catch (Exception e) {
                        e.printStackTrace();
                        throw new MyException("500","t_ex_source_data_info表更新增量值异常",e.getMessage());
                    }
                }
            }
        } catch (MyException e) {
            e.printStackTrace();
        } finally {
            taskInfoService.updateTaskRunningStateById(jobTask.getTaskId(),0);
        }

    }

    /**
    * @Deacription 插入日志
    * @MethodName insertTransformLog
    * @param dto
    * @param startTime
    * @param dataNumber
    * @param dirtyDataNumber
    * @return
    * @Author lijianhua
    * @Date 2023/2/21 16:03
    **/
    private void insertTransformLog(String schedulId,JobTaskInfoQuartz jobTask,TransformInfoDto dto,Date startTime, AtomicInteger dataNumber, AtomicInteger dirtyDataNumber) {
        TExTransformLog transformLog = new TExTransformLog();
        transformLog.setLogId(UUIDGenerator.generator());
        transformLog.setSchedulId(schedulId);
        transformLog.setJobId(jobTask.getJobId());
        transformLog.setTaskId(jobTask.getTaskId());
        transformLog.setTransformId(dto.getTransformId());
        transformLog.setTransformName(dto.getTransformName());
        transformLog.setStartTime(startTime);
        transformLog.setEndTime(new Date());
        transformLog.setReadRows(dataNumber.get());
        transformLog.setWriterRows(dataNumber.get()-dirtyDataNumber.get());
        if(dirtyDataNumber.get()>0){
            transformLog.setExeStaus(0);
        }else{
            transformLog.setExeStaus(1);
        }
        transformLogService.insert(transformLog);
    }

}
